package com.niit.utils

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}


/*
SparkUtil 是对 Spark Context【Spark Core --RDD】
              Spark Session 【Spark SQL  】
              Steaming Context 【Spark Streaming】
              进行封装
 统一的管理，统一的关闭和释放内存，避免重复创建
 */
object SparkUtils {

  //对SparkContext进行统一管理，建立专门存储Spark Context 的 线程池
  private val scLocal = new ThreadLocal[SparkContext]
  //对SparkSession进行统一管理，建立专门存储SparkSession 的 线程池
  private val sparkLocal = new ThreadLocal[SparkSession]
  //对StreamingContext进行统一管理，建立专门存储SparkSession 的 线程池
  private val sscLocal = new ThreadLocal[StreamingContext]

  private var sc:SparkContext = _
  private var spark:SparkSession = _
  private var ssc:StreamingContext = _
  private var seconds:Duration = _

 //统一的创建 SparkContext SparkSession Spark Streaming
  def createSpark(sparkConf:SparkConf,seconds:Duration = Seconds(3)):SparkContext={

    if(sc == null){
      spark = SparkSession.builder().config(sparkConf).getOrCreate()
      sc = spark.sparkContext
      //将创建好的sc 和 spark 添加到 本地线程池进行存储
      putSC(sc)
      putSpark(spark)

      if (this.seconds != seconds){
        //删除之前的Streaming Context
        sscLocal.remove()
        ssc = new StreamingContext(sc,seconds)
        this.seconds = seconds
      }
      putSSC(ssc)
    }
    sc
  }

  def getOrCreateStreamingContext(sparkContext: SparkContext,seconds: Duration):StreamingContext={
    if(this.seconds != seconds){
      sscLocal.remove()
      ssc = new StreamingContext(sc,seconds)
      this.seconds = seconds
      putSSC(ssc)
    }
    ssc
  }


  //将Spark Context添加到本地的线程池的方法
  private def putSC(sc:SparkContext): Unit ={
    scLocal.set(sc)
  }

  //将Spark Session添加到本地的线程池的方法
  private def putSpark(spark:SparkSession): Unit ={
    sparkLocal.set(spark)
  }

  //将Spark Streaming添加到本地的线程池的方法
  private def putSSC(ssc:StreamingContext): Unit ={
    sscLocal.set(ssc)
  }

  //获得已经创建好的Sc spark SSC
  def takeSC():SparkContext={
    scLocal.get()
  }

  def takeSpark():SparkSession={
    sparkLocal.get()
  }

  def takeSSC():StreamingContext={
    sscLocal.get()
  }

  def clear():Unit={
    scLocal.remove()
    sparkLocal.remove()
    sscLocal.remove()
  }

}
